package ru.mipt.victator.pagerank.index;

import java.util.Properties;

import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;

import ru.mipt.victator.pagerank.Value;

public class IndexJob
{
	public static JobConf config(Properties props)
	{
		JobConf conf = new JobConf(IndexJob.class);
		
		conf.setJobName("PageRank-IndexJob");
		
		//map input
		Path pPR = new Path( props.getProperty("OutputData.PageRank.Dir"), "0" );
		conf.setInputPath(pPR);
	    conf.setInputFormat(SequenceFileInputFormat.class);
	    Path pWikiDir = new Path(props.getProperty("InputData.Wiki.Dir"));
	    conf.setWorkingDirectory(pWikiDir);
	    
	    //map output,reduce input
	    conf.setMapOutputKeyClass(Text.class);
	    conf.setMapOutputValueClass(Value.class);
	    
	    //reduce output
	    conf.setOutputFormat(SequenceFileOutputFormat.class);
	    conf.setOutputKeyClass(Text.class);
	    conf.setOutputValueClass(Value.class);
	    Path pIndex = new Path(props.getProperty("OutputData.Index.Dir"));
	    conf.setOutputPath(pIndex);
	    
	    //process options
	    int MapTasks = Integer.parseInt(props.getProperty("IndexJob.MapTasks.Int"));
		conf.setNumMapTasks(MapTasks);
		conf.setMaxMapAttempts(3);
		int ReduceTasks = Integer.parseInt(props.getProperty("IndexJob.ReduceTasks.Int"));
		conf.setNumReduceTasks(ReduceTasks);
		conf.setMaxReduceAttempts(3);
	    
	    //classes options
	    conf.setMapperClass(IndexMap.class);
	    conf.setReducerClass(IndexReduce.class);
		
		return conf;
	}

}
